package com.tn.rtdata.management;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A manager service that monitors heartbeats against ids, firing events if the heartbeats fail.
 */
public class HeartbeatManager
{
  private static final Logger LOGGER = LoggerFactory.getLogger(HeartbeatManager.class);

  private static int threadCount = 0;

  private boolean running;
  private Map<String, Date> heartbeatCache;
  private List<HeartbeatListener> heartbeatListeners;
  private long terminateTimeout;
  private long warningTimeout;

  /**
   * Creates a new <code>HeartbeatManager</code>.
   *
   * @param heartbeatCache   the cache that the heartbeat timestamps are added to.
   * @param warningTimeout   the maximum time allowed between heartbeats before a warning event occurs.
   * @param terminateTimeout the maximum time allowed between heartbeats before a terminate event occurs.
   */
  public HeartbeatManager(Map<String, Date> heartbeatCache, long warningTimeout, long terminateTimeout)
  {
    this.heartbeatCache = heartbeatCache;
    this.warningTimeout = warningTimeout;
    this.terminateTimeout = terminateTimeout;
    this.heartbeatListeners = new ArrayList<HeartbeatListener>();
  }

  /**
   * Returns the running flag.
   */
  public boolean isRunning()
  {
    return running;
  }

  /**
   * Adds the <code>heartbeatListener</code>.
   */
  public void addHeartbeatListener(HeartbeatListener heartbeatListener)
  {
    heartbeatListeners.add(heartbeatListener);
  }

  /**
   * Removes the <code>heartbeatListener</code>.
   */
  public void removeHeartbeatListener(HeartbeatListener heartbeatListener)
  {
    heartbeatListeners.remove(heartbeatListener);
  }

  /**
   * Starts this <code>HeartbeatManager</code>.
   */
  public void start()
  {
    if (!running)
    {
      LOGGER.info("Starting heartbeat monitor.");

      running = true;

      Thread heartbeatMonitorThread = new Thread(new HeartbeatMonitor(), "heartbeat-" + threadCount++);
      heartbeatMonitorThread.setDaemon(true);
      heartbeatMonitorThread.start();
    }
  }

  /**
   * Stops this <code>HeartbeatManager</code>.
   */
  public void stop()
  {
    LOGGER.info("Stopping heartbeat monitor.");
    running = false;
  }

  /**
   * Updates the heartbeat for the <code>id</code>.
   */
  public void heartbeat(String id)
  {
    LOGGER.debug("Heartbeat received: {}", id);
    heartbeatCache.put(id, new Date());
  }

  /**
   * Removes the id from the heartbeat monitor.
   */
  public void remove(String id)
  {
    heartbeatCache.remove(id);
  }

  /**
   * Fires the <code>HeartbeatListener.terminate(String)</code> method for all listeners passing the <code>id</code>.
   */
  private void fireTerminateListeners(String id)
  {
    for (HeartbeatListener heartbeatListener : heartbeatListeners)
    {
      heartbeatListener.terminate(id);
    }
  }

  /**
   * Fires the <code>HeartbeatListener.warning(String)</code> method for all listeners passing the <code>id</code>.
   */
  private void fireWarningListeners(String id)
  {
    for (HeartbeatListener heartbeatListener : heartbeatListeners)
    {
      heartbeatListener.warning(id);
    }
  }

  /**
   * The listener interface used to pass events relating to the heartbeats.
   */
  public static abstract class HeartbeatListener
  {
    /**
     * Invoked when the heartbeat for the <code>id</code> has failed and any associated processing should be
     * terminated.
     */
    @SuppressWarnings("UnusedParameters")
    public void warning(String id)
    {
      //Do nothing, specializations of this class can override if they need to.
    }

    /**
     * Invoked when the heartbeat for the <code>id</code> has failed and any associated processing should be
     * terminated.
     */
    public abstract void terminate(String id);
  }

  /**
   * An implementation of <code<Runnable</code> that checks the heartbeat cache for stale entries, firing events
   * accordingly.
   */
  private class HeartbeatMonitor implements Runnable
  {
    private Set<String> failingSessions = new HashSet<String>();

    /**
     * {@inheritDoc}
     */
    @Override
    public void run()
    {
      while (running)
      {
        try
        {
          Date now = new Date();

          for (String id : new HashSet<String>(heartbeatCache.keySet()))
          {
            Date lastHeartbeat = heartbeatCache.get(id);

            if (after(now, lastHeartbeat, terminateTimeout))
            {
              LOGGER.warn(
                "No heartbeat received for: {} since: {}, terminating",
                new Object[]{id, lastHeartbeat}
              );
              fireTerminateListeners(id);
              failingSessions.remove(id);
            }
            else if (after(now, lastHeartbeat, warningTimeout) && !failingSessions.contains(id))
            {
              LOGGER.warn(
                "No heartbeat received for: {} since: {}, will terminate within: {}ms",
                new Object[]{id, lastHeartbeat, terminateTimeout - warningTimeout}
              );
              fireWarningListeners(id);
              failingSessions.add(id);
            }
            else
            {
              failingSessions.remove(id);
            }
          }

          Thread.sleep(warningTimeout);
        }
        catch (InterruptedException e)
        {
          LOGGER.error("Heartbeat thread interrupted", e);
        }
      }

      threadCount--;
    }

    /**
     * Returns <code>true</code> of the <code>Date now</code> is after the <code>lastHeartbeat + adjustment</code>;
     * otherwise <code>false</code>.
     */
    private boolean after(Date now, Date lastHeartbeat, long adjustment)
    {
      return now.after(new Date(lastHeartbeat.getTime() + adjustment));
    }
  }
}
